package cn.mengfly.testnrz.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.zookeeper.Watcher;
import org.junit.Test;

import java.nio.charset.StandardCharsets;

/**
 * 客户端监听
 */
public class ZkWatcherDemo {

    private String workerPath = "/test/listener/remoteNode";
    private String subWorkerPath = "/test/listener/remoteNode/id-";
    private static final String ZK_ADDRESS = "127.0.0.1:2181";

    @Test
    public void testWatcher() {
        try (final CuratorFramework client = ClientFactory.createSimple(ZK_ADDRESS)) {
            client.start();
            // 判断节点是否存在，没有则创建
            if (!ClientUtil.isNodeExists(client, workerPath)) {
                ClientUtil.createNode(client, workerPath, null);
            }
            Watcher w = watchedEvent -> System.out.println("监听到变化watchedEvent" + watchedEvent);
            byte[] content = client.getData().usingWatcher(w).forPath(workerPath);
            System.out.println("监听节点内容：" + new String(content, StandardCharsets.UTF_8));

            // 第一次变更节点数据
            client.setData().forPath(workerPath, "第一次更改内容".getBytes(StandardCharsets.UTF_8));

            // 第二次更改节点数据
            client.setData().forPath(workerPath, "第二次更改内容".getBytes(StandardCharsets.UTF_8));

            Thread.sleep(Integer.MAX_VALUE);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testNodeCache() {
        try (final CuratorFramework client = ClientFactory.createSimple(ZK_ADDRESS)) {
            client.start();

            NodeCache nodeCache = new NodeCache(client, workerPath, false);
            NodeCacheListener listener = () -> {
                ChildData childData = nodeCache.getCurrentData();
                System.out.println("ZNode节点状态改变, path = " + childData.getPath());
                System.out.println("ZNode节点状态改变, data = " + new String(childData.getData(), StandardCharsets.UTF_8));
                System.out.println("ZNode节点状态改变, stat = " + childData.getStat());
            };
            // 启动节点的事件监听
            nodeCache.getListenable().addListener(listener);
            nodeCache.start();


            // 判断节点是否存在，没有则创建
            if (!ClientUtil.isNodeExists(client, workerPath)) {
                ClientUtil.createNode(client, workerPath, null);
            }
            Thread.sleep(1000);

            // 第一次变更节点数据
            client.setData().forPath(workerPath, "第一次变更节点数据".getBytes(StandardCharsets.UTF_8));
            Thread.sleep(1000);

            // 第二次更改节点数据
            client.setData().forPath(workerPath, "第二次更改节点数据".getBytes(StandardCharsets.UTF_8));
            Thread.sleep(1000);

            // 第三次变更节点数据
            client.setData().forPath(workerPath, "第三次变更节点数据".getBytes(StandardCharsets.UTF_8));
            Thread.sleep(1000);

            // 第四次更改节点数据
            client.delete().forPath(workerPath);
            System.out.println("删除节点");
            Thread.sleep(1000);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testPathChildrenCache() {
        try (final CuratorFramework client = ClientFactory.createSimple(ZK_ADDRESS)) {
            client.start();
            // 判断节点是否存在，没有则创建
            if (!ClientUtil.isNodeExists(client, workerPath)) {
                ClientUtil.createNode(client, workerPath, null);
            }
            PathChildrenCache cache = new PathChildrenCache(client, workerPath, true);
            PathChildrenCacheListener listener = (curatorFramework, event) -> {
                final ChildData data = event.getData();
                switch (event.getType()) {
                    case CHILD_ADDED:
                        System.out.println("子节点增加， path = " + data.getPath() +
                                " data = " + new String(data.getData(), StandardCharsets.UTF_8));
                        break;
                    case CHILD_UPDATED:
                        System.out.println("子节点更新， path = " + data.getPath() +
                                " data = " + new String(data.getData(), StandardCharsets.UTF_8));
                        break;
                    case CHILD_REMOVED:
                        System.out.println("子节点删除， path = " + data.getPath() +
                                " data = " + new String(data.getData(), StandardCharsets.UTF_8));
                        break;
                }
            };

            cache.getListenable().addListener(listener);
            cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            Thread.sleep(1000);
            // 创建3个节点
            for (int i = 0; i < 3; i++) {
                ClientUtil.createNode(client, subWorkerPath + i, null);
            }
            Thread.sleep(1000);
            // 删除三个节点
            for (int i = 0; i < 3; i++) {
                ClientUtil.deleteNode(client, subWorkerPath + i);
            }

        } catch (
                Exception e) {
            e.printStackTrace();
        }

    }
}
